package com.dbsop.rest;

import com.dbsop.domain.EsCondition;
import com.dbsop.domain.EsUpdateCondition;
import com.dbsop.domain.info.EsFiledInfo;
import com.dbsop.domain.params.*;
import com.dbsop.domain.rest.EsRestCondition;
import com.dbsop.domain.transport.EsAggTcCondition;
import com.dbsop.domain.transport.EsTransportCondition;
import com.dbsop.domain.transport.res.XContentBuilderUtils;
import com.dbsop.exception.EsRunException;
import com.dbsop.mapper.IndexMappingInfo;
import com.dbsop.operation.query.rest.RestAggQuerySearch;
import com.dbsop.operation.query.rest.RestQuerySearch;
import com.dbsop.operation.query.transport.TcQuerySearch;
import com.dbsop.request.ClientRequest;
import com.dbsop.request.RestStatus;
import com.dbsop.rest.handler.RestAggHandler;
import com.dbsop.rest.handler.RestRecordHandler;
import com.dbsop.transport.handler.TcRecordHandler;
import com.dbsop.wrapper.DataWrapper;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.reindex.*;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.metrics.InternalValueCount;
import org.elasticsearch.search.aggregations.metrics.ParsedValueCount;
import org.elasticsearch.search.aggregations.metrics.ValueCountAggregationBuilder;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

@Slf4j
public class RestRequest implements ClientRequest {

    protected RestHighLevelClient rc;


    public RestRequest(RestHighLevelClient rc) {
        this.rc = rc;
    }


    /**
     * 索引操作
     * @param index
     * @param idName
     * @param sourceMap
     * @return
     */
    private IndexRequest getIndexRequest(String index, String idName, Map<String, Object> sourceMap){
        IndexRequest indexRequest = new IndexRequest(index);
        if (idName != null){
            Object idValue = sourceMap.get(idName);
            if(idValue == null) throw new EsRunException("ID Value Not Null");
            indexRequest.id(String.valueOf(idValue));
        }
        indexRequest.source(sourceMap);
        return indexRequest;
    }

    @Override
    public <T> List<T> selectList(RequestParams<T> requestParams) {
        EsCondition esCondition  = requestParams.getEsCondition();
        DataWrapper<T> dataWrapper = requestParams.getDataWrapper();
        Class resClass = requestParams.getResClass();

        RestQuerySearch search = new RestQuerySearch((EsRestCondition) esCondition);
        SearchRequest searchRequest = search.getSearchRequest();
        SearchResponse searchResponse = null;
        try {
            Long start = System.currentTimeMillis();
            searchResponse =  rc.search(searchRequest, RequestOptions.DEFAULT);
            Long end =System.currentTimeMillis();
            log.info("调用selectList方法耗时:" +(end-start) +"毫秒");

        } catch (IOException e) {
            throw new RuntimeException(e);
        }
        if (dataWrapper != null){
            return RestRecordHandler.records(searchResponse,null,dataWrapper);
        }
        if (resClass == null || Map.class.isAssignableFrom(resClass)){
            return (List<T>) TcRecordHandler.recordsToMap(searchResponse);
        }
        return TcRecordHandler.records(searchResponse,resClass,null);
    }

    @Override
    public <T> int save(SaveParams<T> saveParams) {
        Map<String, Object> sourceMap = saveParams.getSourceMap();
        String[] indexArr = saveParams.getIndex();
        String idName = saveParams.getIdName();
        IndexRequest indexRequest = getIndexRequest(indexArr[0], idName, sourceMap);
        IndexResponse indexResponse = null;
        try {
            Long start = System.currentTimeMillis();
            indexResponse = rc.index(indexRequest, RequestOptions.DEFAULT);
            Long end =System.currentTimeMillis();
            log.info("调用save方法耗时:" +(end-start) +"毫秒");

        } catch (IOException e) {
            throw new RuntimeException(e);
        }
        if (RestStatus.saveStatus(indexResponse.status().toString()))return 1;
        return 0;
    }

    @Override
    public <T> int saveBatch(SaveParams<T> saveParams) {
        Map<String, List<Map<String, Object>>> mapList = saveParams.getListSourceMap();
        BulkRequest bulkRequest = new BulkRequest();
        String idName = saveParams.getIdName();
        mapList.forEach((index, maps) -> {
            maps.forEach((sourceMap) -> { bulkRequest.add(getIndexRequest(index,idName,sourceMap)); });
        });
        BulkResponse bulkResponse = null;
        try {
            Long start = System.currentTimeMillis();
            bulkResponse = rc.bulk(bulkRequest, RequestOptions.DEFAULT);
            Long end =System.currentTimeMillis();
            log.info("调用saveBatch方法耗时:" +(end-start) +"毫秒");

        } catch (IOException e) {
            throw new RuntimeException(e);
        }
        if (RestStatus.saveStatus(bulkResponse.status().toString()))return mapList.size();
        return 0;
    }

    @Override
    public int updateByQuery(ModifyParams modifyParams) {
        EsUpdateCondition condition = modifyParams.getCondition();
        if (condition.getFieldValue().size() ==0)return 0;
        RestQuerySearch restQuerySearch = new RestQuerySearch(condition.getEsCondition());

        UpdateByQueryRequest updateByQueryRequest = new UpdateByQueryRequest();
        QueryBuilder query = restQuerySearch.getSearchRequest().source().query();
        updateByQueryRequest.setQuery(query);
        updateByQueryRequest.indices(condition.getEsCondition().getIndices());
        updateByQueryRequest.setScript(XContentBuilderUtils.getScript(condition.getFieldValue()));
        BulkByScrollResponse bulkByScrollResponse = null;
        try {
            Long start = System.currentTimeMillis();
            bulkByScrollResponse =  rc.updateByQuery(updateByQueryRequest,RequestOptions.DEFAULT);
            Long end =System.currentTimeMillis();
            log.info("调用updateByQuery方法耗时:" +(end-start) +"毫秒");

        } catch (IOException e) {
            throw new RuntimeException(e);
        }
        BulkByScrollTask.Status status = bulkByScrollResponse.getStatus();
        return (int) status.getUpdated();
    }

    @Override
    public int deleteBySearch(DeleteParams deleteParams) {
        EsCondition condition = deleteParams.getEsCondition();
        TcQuerySearch transportCondition = new TcQuerySearch((EsTransportCondition) condition);
        DeleteByQueryRequest deleteByQueryRequest = new DeleteByQueryRequest();
        QueryBuilder query = transportCondition.getSearchRequest().source().query();
        deleteByQueryRequest.indices(condition.getIndices());
        deleteByQueryRequest.setQuery(query);
        BulkByScrollResponse bulkByScrollResponse = null;
        try {

            Long start = System.currentTimeMillis();
            rc.deleteByQuery(deleteByQueryRequest,RequestOptions.DEFAULT);
            Long end =System.currentTimeMillis();
            log.info("调用deleteBySearch方法耗时:" +(end-start) +"毫秒");

        } catch (IOException e) {
            throw new RuntimeException(e);
        }
        BulkByScrollTask.Status status = bulkByScrollResponse.getStatus();
        return (int) status.getDeleted();
    }

    @Override
    public <T> List<T> selectListAgg(AggRequestParams<T> aggRequestParams) {
        EsAggTcCondition esAggCondition = (EsAggTcCondition) aggRequestParams.getEsAggCondition();
        Class<T> resClass = aggRequestParams.getResClass();
        DataWrapper<T> dataWrapper = aggRequestParams.getDataWrapper();
        esAggCondition.recordSize(0);
        RestAggQuerySearch aggQuerySearch = new RestAggQuerySearch(esAggCondition);

        SearchResponse searchResponse = null;
        try {
            Long start = System.currentTimeMillis();
            searchResponse = rc.search(aggQuerySearch.getSearchRequest(), RequestOptions.DEFAULT);
            Long end =System.currentTimeMillis();
            log.info("调用selectListAgg方法耗时:" +(end-start) +"毫秒");
        } catch (IOException e) {
            e.printStackTrace();
        }
        List<Map<String, Object>> mapList = RestAggHandler.parseAgg(searchResponse);
        List<T> datas = new ArrayList<T>();
        if (mapList == null || mapList.size() == 0){
            return datas;
        }
        if (dataWrapper!=null ){
            for (Map<String, Object> objectMap : mapList) {
                T wrapper = dataWrapper.wrapper(objectMap);
                if (wrapper !=null){
                    datas.add(wrapper);
                }
            }
            return datas;
        }
        if (resClass == Map.class){
            return (List<T>) mapList;
        }

        return datas;
    }

    @Override
    public <E> List<E> collapseList(CollapseParams<E> collapseParams) {
        DataWrapper<E> dataWrapper = collapseParams.getDataWrapper();
        EsCondition esCondition = collapseParams.getEsCondition();
        Class<E> resClass = collapseParams.getResClass();
        if (esCondition.getRanking() == null){
            throw new RuntimeException("排名条件未设置");
        }
        SearchResponse searchResponse = null;
        try {
            Long start = System.currentTimeMillis();
            searchResponse = rc.search(new RestQuerySearch(esCondition).getSearchRequest(), RequestOptions.DEFAULT);
            Long end =System.currentTimeMillis();
            log.info("调用collapseList方法耗时:" +(end-start) +"毫秒");
        } catch (IOException e) {
            e.printStackTrace();
        }
        return RestRecordHandler.rankingRecord(searchResponse,resClass,dataWrapper);
    }

    @Override
    public <E> ESPage<E> selectPageList(PageParams<E> pageParams) {
        DataWrapper<E> dataWrapper = pageParams.getDataWrapper();
        EsCondition esCondition = pageParams.getEsCondition();
        Class<E> resClass = pageParams.getResClass();
        ESPage page = pageParams.getPage();

        RestQuerySearch querySearch = new RestQuerySearch(esCondition);
        SearchRequest searchRequest = querySearch.getSearchRequest();
        ValueCountAggregationBuilder count = AggregationBuilders.count("count").field("_id");
        searchRequest.source().aggregation(count);
        SearchResponse searchResponse = null;
        try {
            Long start = System.currentTimeMillis();
            searchResponse = rc.search(searchRequest, RequestOptions.DEFAULT);
            Long end =System.currentTimeMillis();
            log.info("调用selectPageList方法耗时:" +(end-start) +"毫秒");
        } catch (IOException e) {
            e.printStackTrace();
        }
        long totalSize = ((ParsedValueCount) searchResponse.getAggregations().asList().get(0)).getValue();
        List<E> tList = TcRecordHandler.records(searchResponse,resClass,dataWrapper);
        page.setContent(tList);
        page.setTotalSize(totalSize);
        return page;
    }

}
